package com.telra.app.server.jms;

import com.telra.app.server.data.OrderEvent;
import com.telra.app.server.protobuff.OrderProtos;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;

import javax.jms.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


/**
 * Simulates order traffic by randomly creating orders and publishing them on JMS.
 * Can be throttled to increase/decrease rate to testing load.
 *
 * @author jamesarlet
 *
 */
public class Publisher {

    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(Publisher.class);

    public Publisher() {

        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
        executor.scheduleAtFixedRate(new PublisherThread(), 10, 10, TimeUnit.MILLISECONDS);

    }

    private class PublisherThread implements Runnable {

        Session session = null;
        MessageProducer producer = null;

        public PublisherThread() {

            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            connectionFactory.setUseAsyncSend(true);

            try {
                Connection connection = connectionFactory.createConnection();
                connection.start();

                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

                Topic topic = session.createTopic("TEST.OUT");
                producer = session.createProducer(topic);
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                log.info("started publisher");
            }
            catch (JMSException e) {
                e.printStackTrace();
            }

        }

        public void run() {

            try {
                ObjectMessage bMessage = session.createObjectMessage();
                OrderProtos.OrderTicket o = OrderEvent.generateOrder();
                bMessage.setObject(o);
                producer.send(bMessage);

                log.info("sent: " + o.getItemName() + "," + o.getItemCategory() + "," + o.getPrice());
            }

            catch (JMSException e) {
                e.printStackTrace();
            }

        }

    }
}